//
// Created by yinlijun on 2017/9/30.
// 服务器线程池，维护线程池和工作队列。
//
#include "../utils/app_sys_log.h"
#include <pthread.h>
#include <errno.h>
#include <sys/types.h>
#include <unistd.h>

#include "../works/work_data.h"
#include "server_thread_pool.h"
#include "../works/work_item_queue.h"
#include "../common/rocky_connection.h"

struct ServerThreadPool {
    WorkItemQueue* pending_item_queue;
    WorkItemQueue* working_item_queue;
    int pool_size;
    int shutdown;
    pthread_t *work_threads;
    pthread_mutex_t locker;
    pthread_cond_t ready_cond;
};

static void my_work_data_destroy_func(void* data) {
    work_data_destroy((WorkData *)data);
}

void* thread_pool_worker(void* arg) {
    ServerThreadPool* thiz = (ServerThreadPool*)arg;
    while(1) {
        pthread_mutex_lock(&thiz->locker);

        WorkData *work_data;
        while (!thiz->shutdown &&
                (work_data = (WorkData *)work_item_queue_remove(thiz->pending_item_queue)) == NULL) {
            //等待数据或线程池关闭
            pthread_cond_wait(&thiz->ready_cond, &thiz->locker);
        }
        if(thiz->shutdown) {
            pthread_mutex_unlock(&thiz->locker);
            app_sys_warn("pthread %p exit", pthread_self());
            pthread_exit(NULL);
        }
        work_item_queue_add(thiz->working_item_queue, work_data, my_work_data_destroy_func);
        pthread_mutex_unlock(&thiz->locker);
        app_sys_warn("pthread %p start work", pthread_self());
        work_data_handle(work_data);
    }

    return NULL;
}

static void server_thread_pool_create_threads(ServerThreadPool* thiz, int count) {
    thiz->work_threads = (pthread_t *)malloc(count * sizeof(pthread_t));
    for(int i = 0; i < count; i++) {
        pthread_create(&thiz->work_threads[i], NULL, thread_pool_worker, thiz);
    }
}

ServerThreadPool* server_thread_pool_add(ServerThreadPool* thiz, RockyConnection* connection) {
    WorkData *work_data = work_data_create(connection, rocky_connectoin_handler_callback_for_server_thread,
                                           rocky_connection_destroy_callback_for_server_thread, thiz);
    work_item_queue_add(thiz->pending_item_queue, work_data, my_work_data_destroy_func);
    pthread_cond_signal(&thiz->ready_cond);

    return thiz;
}

ServerThreadPool* server_thread_pool_create(int size) {
    if(size <= 0) {
        app_sys_error("thread pool size must be bigger than 0");
        return NULL;
    }
    ServerThreadPool* thiz = (ServerThreadPool *)malloc(sizeof(ServerThreadPool));
    if(thiz == NULL) {
        app_sys_error("thread pool create failed: %s", strerror(errno));
        return NULL;
    }

    pthread_mutex_init(&thiz->locker, NULL);
    pthread_cond_init(&thiz->ready_cond, NULL);
    thiz->pool_size = size;
    thiz->shutdown = 0;
    thiz->pending_item_queue = work_item_queue_create();
    thiz->working_item_queue = work_item_queue_create();
    server_thread_pool_create_threads(thiz, size);

    return thiz;
}

static void server_thread_pool_join_all_threads(ServerThreadPool* thiz) {
    pthread_cond_broadcast(&thiz->ready_cond);
    for (int i = 0; i < thiz->pool_size; i++) {
        pthread_join(thiz->work_threads[i], 0);
    }
    free(thiz->work_threads);
}

static void work_data_shutdown_func(void* data, void* ctx) {
    WorkData* work_data = (WorkData *)data;
    RockyConnection *connection = (RockyConnection *)work_data_get_data(work_data);
    rocky_connection_shutdown(connection);
}

void server_thread_pool_destroy(ServerThreadPool* thiz) {
    if(thiz == NULL) {
        return;
    }
    thiz->shutdown = 1;
    //关闭所有连接
    work_item_queue_foreach(thiz->pending_item_queue, work_data_shutdown_func, NULL);
    work_item_queue_foreach(thiz->working_item_queue, work_data_shutdown_func, NULL);
    server_thread_pool_join_all_threads(thiz);
    work_item_queue_destroy(thiz->pending_item_queue);
    work_item_queue_destroy(thiz->working_item_queue);

    free(thiz);
}

#ifdef __TEST_SERVER_THREAD_POOL__
void* my_work_handler(void* data, void* ctx) {
    app_sys_debug("work handler data: %d", data);
    return NULL;
}
int main() {
    ServerThreadPool *thread_pool = server_thread_pool_create(10);
    int data_size = 200;
    WorkData** work_data_array = (WorkData **) malloc(data_size * sizeof(WorkData *));
    for (int i = 0; i < data_size; i++) {
        usleep(10 * 1000);
        work_data_array[i] = work_data_create((void *)i, my_work_handler, NULL, NULL);
        server_thread_pool_add(thread_pool, work_data_array[i]);
    }

    sleep(3);
    server_thread_pool_destroy(thread_pool);
}
#endif